package com.seal.commons.infrastructure.rocketmq.listener;

import com.seal.commons.infrastructure.facade.jms.MessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.Map;

/**
 * ClassName:DefaultMessageListener <br/>
 * Date:     2019-06-06 11:00 <br/>
 *
 * @author yong.huang
 * @see
 * @since JDK 1.8
 */

@Slf4j
public class DefaultMessageListener implements MessageListenerConcurrently {

    private Map<String, MessageListener> listenerMap;

    public DefaultMessageListener(Map<String, MessageListener> listenerMap){
        this.listenerMap = listenerMap;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        MessageExt messageExt = list.get(0);
        log.info("DefaultMessageListener on message ,topicName:[{}] msgId:[{}]",messageExt.getTopic(),messageExt.getMsgId());
        MessageListener listener = listenerMap.get(messageExt.getTopic() + "_" + messageExt.getTags());
        String msgId = messageExt.getMsgId();
        String tag = messageExt.getTags();
        try {
            //TODO 是否需要根据TAGS进行消息过滤
            listener.onMessage(msgId, tag, messageExt.getBody());
        } catch (Throwable e){
            log.error(" msgId:[{"+msgId+"}] tag:[{"+tag+"}] error ",e);
            throw e;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
